package cn.azzhu.proj

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.io.Source

/**
 * @author azzhu
 * @create 2020-09-25 23:33:20
 */
object KafkaProduceUserBehaviourData {
  def main(args: Array[String]): Unit = {
    writeToKafka("hotitems")
  }

  def writeToKafka(topic: String):Unit = {
    val prop = new Properties()
    prop.setProperty("bootstrap.servers","hadoop105:9092,hadoop106:9092,hadoop107:9092")
    prop.setProperty("key.serializer","org.apache.kafka.common.serialization.StringSerializer")
    prop.setProperty("value.serializer","org.apache.kafka.common.serialization.StringSerializer")

    val producer = new KafkaProducer[String,String](prop)
    val bufferedSource = Source.fromFile("D:\\bigdata\\flink-learning\\src\\main\\resources\\UserBehavior.csv")

    for(line <- bufferedSource.getLines()) {
      val record = new ProducerRecord[String,String](topic,line)
      producer.send(record)
    }
    producer.close()
  }
}
